package com.gitee.xhs.keyedState.valueState;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.configuration.Configuration;

/**
 * @author BigHead
 * @date 2022/6/29 13:57
 * @description ValueStateKeyedProcessFunction
 */
public class ValueStateKeyedProcessFunction extends KeyedProcessFunction<Long, Transaction, String> {

    private transient ValueState<Integer> valueState;

    @Override
    public void open(Configuration configuration) throws Exception {
        valueState = getRuntimeContext().getState(
                ValueStateConstant.VALUE_STATE_DESCRIPTOR
        );
    }

    @Override
    public void processElement(Transaction value, KeyedProcessFunction<Long, Transaction, String>.Context ctx, Collector<String> out) throws Exception {
        Integer stateInteger = 0;

        if (null != valueState.value()) {
            stateInteger = valueState.value();
        }
        stateInteger += 1;
        valueState.update(stateInteger);

        out.collect(value.getAccountId() + " 计数 =>" + stateInteger);
    }
}
